package com.leiyuee.flink.batch.job;

import com.leiyuee.flink.batch.util.DataFormatUtils;
import com.leiyuee.flink.batch.util.FileUtil;
import com.leiyuee.flink.tools.SinkBuilder;
import com.leiyuee.flink.tools.constants.MySQLConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;

/**
 * 读取 Txt文件内容，导入Doris: 60分钟1次
 *  优化：查前2小时的数据，去掉已导入的数据，按照时间排序，
 * @author Yu_Lei
 * @date 2024/6/27
 */
@Slf4j
public class MovementJob {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public static void main(String[] args) throws Exception {
        System.out.println("读取Txt文件内容，导入Doris");

        String filePath = "/usr/local/app/cnhuam0sfmmapp1/SAPData/CN02_Movements_ALL/AWS WMS";
        // 读取文件内容回去最近一小时的文件
        List<File> filesByDate = FileUtil.getRecentNHoursFiles(filePath,1);
        if (filesByDate.isEmpty()) {
            throw new RuntimeException("导入失败，未读取到文件，请核查：");
        }
        List<RowData> all = new ArrayList<RowData>();
        try {
            for (File file : filesByDate) {
                System.out.println("读取文件：" + file.getName());
                // 读取文件内容
//                File file = new File(filePath);
                FileInputStream fileInputStream = new FileInputStream(file);
                InputStreamReader reader = new InputStreamReader(fileInputStream, "UTF-8");
                // 解析文件内容
                BufferedReader bufferedReader = new BufferedReader(reader);
                List<RowData> rowData = DataFormatUtils.praseTxtToRowDataBySort(bufferedReader,60);
                all.addAll(rowData);
                reader.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("导入失败，数据读取异常！");
        }
        if (all.isEmpty()){
            throw new RuntimeException("导入失败，未读取到数据");
        }
//        List<RowData> all = DataFormatUtils.readTextFieldBySort(filePath, 60);
        System.out.println(all.size());
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<RowData> dataStream = env.fromCollection(all);
        //         Doris 配置 Sink To Doris: http://cnhuam0itprd02/8030
        String mysqlDorisDB = "Trace_SAPPO";
        String mysqlDorisTB = "tb_sap_movement";
        String mysqlDorisURL = MySQLConstant.DORIS_PRD02_URL + mysqlDorisDB;
        String mysqlDorisStr = String.format("INSERT INTO %s.%s ", mysqlDorisDB, mysqlDorisTB) +
                " (  material_document,  postng_date,  entered,  gr_gi_slip,  `reference`,  document_date,  `time`,  user_name,  document_header_text,  bill_of_lading,  material,  company_code,  plant,\n" +
                "  item1,  storage_location,  customer,  sales_order1,  opun,  `order`,  movement_type,  unit,  currency,  receive_plant,  mpn_material,  profit_center,\n" +
                "  stge_type,  receive_sloc,  cost_center,  `text`,  purchasing_document,  item2,  qty_in_une,  `entr_ unit`,  batch,  stock_type,  special_stock,  vendor,\n" +
                "  quantity,  d_c_indic,  amount_lc,  mat_year,  sales_order2,  so_item,  storagebin,  ref_document,  reason_for_movement,  reason,  ext_material_group,  del_ind,\n" +
                "  material_type,  material_group,  bulk_matl,  purchasing_group,  gr_pr_time,  material_description,  standard_price,  p_ut,  vendor_name,  lc2_amount)" +
                " VALUES(?,?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?  " +
                "      , ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, " +
                "        ? , ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?);";
        String indexDorisField = String.format("[%s]%s.%s %s", "Doris MySQL", mysqlDorisDB, mysqlDorisTB, "site");
        DataStreamSink<RowData> rowDataDataStreamSink = dataStream.addSink(SinkBuilder.JdbcSink(mysqlDorisStr,
                (statement, event) -> {
                    for (int i = 0;i < 59;i++){
                        statement.setString(i+1, String.valueOf(event.getString(i)));
                    }
                }
                , mysqlDorisURL, MySQLConstant.DORIS_PRD02_USER_JABIL, MySQLConstant.DORIS_PRD02_USER_JABIL_PWD));

        JobExecutionResult execute = env.execute("MovementJob:"+Timestamp.from(Instant.now()).toString() + " 数据量："+all.size());
    }


}
